package com.xrui.hbase.codec.impl;

import com.google.common.base.Preconditions;

import com.google.common.collect.*;
import com.xrui.hbase.*;
import com.xrui.hbase.codec.CellDecoder;
import com.xrui.hbase.codec.CellDecoderFactory;
import com.xrui.hbase.exception.DecoderNotFoundException;
import com.xrui.hbase.impl.BoundColumnReaderSpec;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.schema.spec.CellSpec;
import com.xrui.hbase.schema.spec.ColumnReaderSpec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Provider for cell decoders of a given table.
 * <br>
 * <p> HBaseCell decoders for all columns in the table are pro-actively created when the
 * CellDecoderProvider is constructed. HBaseCell decoders are cached and reused.</p>
 * <p> At construction time, cell decoders may be customized by specifying BoundColumnReaderSpec
 * instances to overlay on top of the actual table layout, using the constructor:
 * {@link #create(TableLayout, Map, Collection, TableReaderBuilder.OnDecoderCacheMiss)}.</p>
 * CellSpec customizations include:
 * <ul>
 * <li> choosing between generic and specific Avro records. </li>
 * <li> choosing a different Avro reader schema. </li>
 * <li> using the Avro writer schema (this forces using generic records). </li>
 * </ul>
 * <h2>Thread Safety</h2>
 * {@code CellDecoderProvider} is not thread safe.
 */
public final class CellDecoderProvider {
    private static final Logger LOG = LoggerFactory.getLogger(CellDecoderProvider.class);

    /**
     * Layout of the table for which decoders are provided.
     */
    private final TableLayout mLayout;

    /**
     * HBaseCell decoders for columns, including data request and table reader overrides.
     */
    private final ImmutableMap<ColumnName, CellDecoder<?>> mColumnDecoders;

    /**
     * HBaseCell decoders for alternative columns specified in
     * {@link #create(TableLayout, Map, Collection, TableReaderBuilder.OnDecoderCacheMiss)}.
     */
    private final Map<BoundColumnReaderSpec, CellDecoder<?>> mReaderSpecDecoders;

    /**
     * Behavior when a decoder cannot be found.
     */
    private final TableReaderBuilder.OnDecoderCacheMiss mOnDecoderCacheMiss;

    /**
     * Constructor for {@code CellDecoderProvider}s.
     *
     * @param layout             {@code TableLayout} of the cell decoder provider's table.
     * @param columnDecoders     Individual cell decoders for the columns of the table.
     * @param readerSpecDecoders Alternative cell decoders for columns of the table.
     * @param onDecoderCacheMiss Determines behavior when a decoder cannot be found.
     */
    private CellDecoderProvider(
        final TableLayout layout,
        final ImmutableMap<ColumnName, CellDecoder<?>> columnDecoders,
        final Map<BoundColumnReaderSpec, CellDecoder<?>> readerSpecDecoders,
        final TableReaderBuilder.OnDecoderCacheMiss onDecoderCacheMiss
    ) {
        mLayout = layout;
        mColumnDecoders = columnDecoders;
        mReaderSpecDecoders = readerSpecDecoders;
        mOnDecoderCacheMiss = onDecoderCacheMiss;
    }

    /**
     * Initialize a provider for cell decoders.
     *
     * @param layout      the layout for which to provide decoders.
     * @param schemaResolver the schema resolver from which to resolve cell schemas.
     * @param factory     Default factory for cell decoders.
     * @param overrides   Column specification overlay/override map.
     *                    Specifications from this map override the actual specification from the table.
     * @return a new {@code CellDecoderProvider}.
     * @throws IOException in case of an error creating the cached decoders.
     */
    public static CellDecoderProvider create(
        final TableLayout layout,
        final AvroSchemaResolver schemaResolver,
        final CellDecoderFactory factory,
        final Map<ColumnName, CellSpec> overrides
    ) throws IOException {
        // Note: nothing prevents one from overriding the specification for one specific qualifier
        // in a map-type family.
        final Set<ColumnName> columns =
            Sets.newHashSet(Iterables.concat(overrides.keySet(), layout.getColumns()));

        // Pro-actively build cell decoders for all columns in the table:
        final ImmutableMap.Builder<ColumnName, CellDecoder<?>> decoderMap =
            ImmutableMap.builder();
        for (ColumnName column : columns) {
            // Gets the specification for this column,
            // from the overlay map or else from the actual table layout:
            CellSpec cellSpec = overrides.get(column);
            if (null == cellSpec) {
                cellSpec = layout.getCellSpec(column);
            } else {
                // Deep-copy the user-provided CellSpec:
                cellSpec = CellSpec.copy(cellSpec);
            }

            // Fills in the missing details to build the decoder:
            if (cellSpec.getSchemaResolver() == null) {
                cellSpec.setSchemaResolver(schemaResolver);
            }
            if (cellSpec.getDecoderFactory() == null) {
                cellSpec.setDecoderFactory(factory);
            }

            final CellDecoder<?> decoder = cellSpec.getDecoderFactory().create(cellSpec);
            decoderMap.put(column, decoder);
        }
        return new CellDecoderProvider(
            layout,
            decoderMap.build(),
            Maps.<BoundColumnReaderSpec, CellDecoder<?>>newHashMap(),
            TableReaderBuilder.DEFAULT_CACHE_MISS);
    }

    /**
     * Create a provider for cell decoders.
     *
     * @param layout             the layout for which to provide decoders.
     * @param overrides          Column specification overlay/override map. Specifications from this map
     *                           override the actual specification from the table.
     * @param onDecoderCacheMiss behavior to use when a decoder cannot be found.
     * @param alternatives       alternate column specifications for which decoders should be provided.
     * @return a new {@code CellDecoderProvider}.
     * @throws IOException in case of an error creating the cached decoders.
     */
    public static CellDecoderProvider create(
        final TableLayout layout,
        final Map<ColumnName, BoundColumnReaderSpec> overrides,
        final Collection<BoundColumnReaderSpec> alternatives,
        final TableReaderBuilder.OnDecoderCacheMiss onDecoderCacheMiss
    ) throws IOException {
        // Pro-actively build cell decoders for all columns in the table and spec overrides:
        return new CellDecoderProvider(
            layout,
            makeColumnDecoderMap(layout, overrides),
            makeSpecDecoderMap(layout, overrides.values(), alternatives),
            onDecoderCacheMiss);
    }

    /**
     * Create a new {@link CellDecoder} from a {@link BoundColumnReaderSpec}.
     *
     * @param layout {@code TableLayout} from which storage information will be retrieved to build
     *               decoders.
     * @param spec   specification of column read properties from which to build a cell decoder.
     * @return a new cell decoder based on the specification.
     * @throws IOException in case of an error making the decoder.
     */
    private static CellDecoder<?> createDecoderFromSpec(
        final TableLayout layout,
        final BoundColumnReaderSpec spec
    ) throws IOException {
        final ColumnReaderSpec.AvroDecoderType decoderType = spec.getColumnReaderSpec().getAvroDecoderType();
        if (null != decoderType) {
            switch (decoderType) {
                case GENERIC: {
                    return DefaultCellDecoderFactory.get().create(layout, spec);
                }
                case SPECIFIC: {
                    return SpecificCellDecoderFactory.get().create(layout, spec);
                }
                default:
                    throw new InternalError("Unknown decoder type: " + decoderType);
            }
        } else {
            // If the decoder type is null, we can use the generic factory.
            return DefaultCellDecoderFactory.get().create(layout, spec);
        }
    }

    /**
     * Build a map of {@link BoundColumnReaderSpec} to {@link CellDecoder} from a collection of
     * specs.
     * <p>
     * All columns in overrides and alternatives are assumed to be included in the table layout
     * because of prior validation.
     *
     * @param layout       TableLayout from which storage information will be retrieved to build
     *                     decoders.
     * @param overrides    specifications of column read properties from which to build decoders.
     * @param alternatives further specifications of column reader properties from which to build
     *                     decoders.
     * @return a map from specification to decoders which follow those specifications.
     * @throws IOException in case of an error making decoders.
     */
    private static Map<BoundColumnReaderSpec, CellDecoder<?>> makeSpecDecoderMap(
        final TableLayout layout,
        final Collection<BoundColumnReaderSpec> overrides,
        final Collection<BoundColumnReaderSpec> alternatives
    ) throws IOException {
        final Map<BoundColumnReaderSpec, CellDecoder<?>> decoderMap = Maps.newHashMap();
        for (BoundColumnReaderSpec spec : overrides) {
            Preconditions.checkState(null == decoderMap.put(spec, createDecoderFromSpec(layout, spec)));
        }
        for (BoundColumnReaderSpec spec : alternatives) {
            Preconditions.checkState(null == decoderMap.put(spec, createDecoderFromSpec(layout, spec)));
        }

        return decoderMap;
    }

    /**
     * Build a map of column names to {@link CellDecoder} for all columns in a given layout and
     * set of overrides.
     * <p>
     * All columns in overrides are assumed to be included in the table layout because of prior
     * validation.
     *
     * @param layout    layout from which to create column names and column specifications.
     * @param overrides overridden column read properties.
     * @return a map from all columns in a table and overrides to decoders for those columns.
     * @throws IOException in case of an error making decoders.
     */
    private static ImmutableMap<ColumnName, CellDecoder<?>> makeColumnDecoderMap(
        final TableLayout layout,
        final Map<ColumnName, BoundColumnReaderSpec> overrides
    ) throws IOException {
        final Set<ColumnName> columns = layout.getColumns();
        final ImmutableMap.Builder<ColumnName, CellDecoder<?>> decoderMap =
            ImmutableMap.builder();
        for (ColumnName column : columns) {
            // Gets the specification for this column,
            // from the overlay map or else from the actual table layout:
            final BoundColumnReaderSpec spec = overrides.get(column);
            if (null != spec) {
                decoderMap.put(column, createDecoderFromSpec(layout, spec));
            } else {
                final CellSpec cellSpec = layout.getCellSpec(column);
                decoderMap.put(column, cellSpec.getDecoderFactory().create(cellSpec));
            }
        }
        return decoderMap.build();
    }

    /**
     * Get a decoder from a {@link BoundColumnReaderSpec}. Creates a new decoder if one does not
     * already exist.
     *
     * @param spec specification of column read properties from which to create a decoder.
     * @param <T>  the type of the value encoded in the cell.
     * @return a new or cached cell decoder corresponding to the given specification.
     * @throws IOException in case of an error create a new decoder.
     */
    @SuppressWarnings("unchecked")
    private <T> CellDecoder<T> getDecoder(BoundColumnReaderSpec spec) throws IOException {
        final CellDecoder<T> decoder = (CellDecoder<T>) mReaderSpecDecoders.get(spec);
        if (null != decoder) {
            return decoder;
        } else {
            switch (mOnDecoderCacheMiss) {
                case FAIL: {
                    throw new DecoderNotFoundException(
                        "Could not find cell decoder for BoundColumnReaderSpec: " + spec);
                }
                case BUILD_AND_CACHE: {
                    LOG.debug(
                        "Building and caching new cell decoder from ColumnReaderSpec: {} for column: {}",
                        spec.getColumnReaderSpec(), spec.getColumn());
                    final CellDecoder<T> newDecoder =
                        (CellDecoder<T>) createDecoderFromSpec(mLayout, spec);
                    mReaderSpecDecoders.put(spec, newDecoder);
                    return newDecoder;
                }
                case BUILD_DO_NOT_CACHE: {
                    LOG.debug(
                        "Building and not caching new cell decoder from ColumnReaderSpec: {} for column: {}",
                        spec.getColumnReaderSpec(), spec.getColumn());
                    return (CellDecoder<T>) createDecoderFromSpec(mLayout, spec);
                }
                default: {
                    throw new InternalError("Unknown OnDecoderCacheMiss: " + mOnDecoderCacheMiss);
                }
            }
        }
    }

    // -----------------------------------------------------------------------------------------------
    // Public interface

    /**
     * Get a {@code CellDecoderProvider} with overrides applied from the provided request.
     *
     * @param request to overlay overrides from.
     * @return a {@code CellDecoderProvider} for the provided request.
     * @throws IOException on unrecoverable IO error.
     */
    public CellDecoderProvider getDecoderProviderForRequest(
        final DataRequest request
    ) throws IOException {
        final List<BoundColumnReaderSpec> readerSpecs = Lists.newArrayList();
        for (DataRequest.Column columnRequest : request.getColumns()) {
            final ColumnReaderSpec readerSpec = columnRequest.getReaderSpec();
            if (readerSpec != null) {
                final ColumnName column = columnRequest.getColumnName();
                readerSpecs.add(BoundColumnReaderSpec.create(readerSpec, column));
            }
        }

        if (readerSpecs.isEmpty()) {
            return this;
        }

        final Map<ColumnName, CellDecoder<?>> columnDecoders = Maps.newHashMap();
        columnDecoders.putAll(this.mColumnDecoders);

        for (BoundColumnReaderSpec readerSpec : readerSpecs) {
            columnDecoders.put(readerSpec.getColumn(), getDecoder(readerSpec));
        }

        return new CellDecoderProvider(
            mLayout,
            ImmutableMap.copyOf(columnDecoders),
            mReaderSpecDecoders,
            mOnDecoderCacheMiss);
    }

    /**
     * Gets a cell decoder for the specified column or (map-type) family.
     * <p>
     * <p>
     * When requesting a decoder for a column within a map-type family, the decoder for the
     * entire map-type family will be returned unless an override has been specified for the
     * exact fully-qualified column.
     * </p>
     *
     * @param column to look up.
     * @param <T>    Type of the data to decode.
     * @return a cell decoder for the specified column. {@code null} if the column does not exist or
     * if the family is not map-type.
     */
    @SuppressWarnings("unchecked")
    public <T> CellDecoder<T> getDecoder(final ColumnName column) {
        final CellDecoder<T> decoder = (CellDecoder<T>) mColumnDecoders.get(column);
        if (decoder != null) {
            // There already exists a decoder for this column:
            return decoder;
        }

        if (column.isFullyQualified()) {
            // There is no decoder for the specified fully-qualified column.
            // Try the family (this will only work for map-type families):
            return getDecoder(ColumnName.create(column.getFamily(), null));
        }

        return null;
    }
}
